/**
 * Copyright (C) 2012-2013 jumei, Inc.
 * ServerSideConnectionManager.java
 * 2015年6月26日
 */
package org.goface.rpc.net;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFactory;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ServerChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;

import org.goface.rpc.io.NettyUtils;
import org.goface.rpc.io.RpcServerSideHandler;
import org.goface.rpc.pipeline.ServicePipeLine;
import org.goface.rpc.pipeline.ServicePipeLineFactory;

import com.google.common.net.HostAndPort;

/**
 * 不同的服务可以使用不同的服务端口
 *
 * @author jm110
 */
public class ServerSideConnectionManager {
    private static ServerSideConnectionManager connectionMgr = new ServerSideConnectionManager();
    private Map<HostAndPort, ServerBootstrap> serverCache = new ConcurrentHashMap<>();
    private Map<HostAndPort, Set<Channel>> clientChannelCache = new ConcurrentHashMap<>();

    public static ServerSideConnectionManager getInstance() {
        return connectionMgr;
    }

    // 同一个address 只会启动一个server 实例
    public ServerBootstrap getServerBootstrap(HostAndPort bindAddress) {
        return serverCache.get(bindAddress);
    }

    public void putServerBootstrap(HostAndPort bindAddress, ServerBootstrap server) {
        serverCache.put(bindAddress, server);
    }

    // 返回连接到制定server 的所有client channel
    public Set<Channel> getClientChannels(HostAndPort bindAddress) {
        return clientChannelCache.get(bindAddress);
    }

    public void addClientChannel(HostAndPort bindAddress, Channel clientChannel) {
        Set<Channel> clientChannels = clientChannelCache.get(bindAddress);
        if (clientChannels == null) {
            clientChannels = new HashSet<Channel>();
            clientChannelCache.put(bindAddress, clientChannels);
        }
        clientChannels.add(clientChannel);
    }

    public void removeClientChannel(HostAndPort bindAddress, Channel clientChannel) {
        Set<Channel> clientChannels = clientChannelCache.get(bindAddress);
        if (clientChannels != null) {
            clientChannels.remove(clientChannel);
        }
    }

    public void startServer(HostAndPort hostAndPort) {
        if (serverCache.get(hostAndPort) != null) {
            // 已经启动过了
            return;
        }
        ServerBootstrap server = new ServerBootstrap();
        SocketAddress address = new InetSocketAddress(hostAndPort.getHostText(), hostAndPort.getPort());
        server.group(new NioEventLoopGroup(1, Executors.newFixedThreadPool(1)),
                new NioEventLoopGroup(3, Executors.newFixedThreadPool(3)));
        server.channelFactory(new ChannelFactory<ServerChannel>() {
            @Override
            public ServerChannel newChannel() {
                return new NioServerSocketChannel();
            }
        });
        final ServicePipeLine pipeLine = ServicePipeLineFactory.createServerSidePipeLine();
        server.childHandler(new ChannelInitializer<Channel>() {
            @Override
            protected void initChannel(Channel ch) throws Exception {
                NettyUtils.addJavaSerialiseCodec(ch, new RpcServerSideHandler(pipeLine));
            };
        });
        try {
            ChannelFuture future = server.bind(address).sync();
            future.get();
            putServerBootstrap(hostAndPort, server);
        } catch (Exception e) {
            if (server != null) {
                try {
                    server.childGroup().close();
                    server.group().close();
                } catch (Exception e1) {

                }

            }
            throw new RuntimeException(e);
        }
    }
}
